package com.galengao.rabbitmq.helloworld;

import com.galengao.rabbitmq.util.RabbitmqUtils;
import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;


/**
 * 消息消费者
 */
@Slf4j
public class MessageConsumer {

    public static void main(String[] args) {
        ConnectionFactory connectionFactory = RabbitmqUtils.connectionFactory();
        try(
                Connection connection= connectionFactory.newConnection();
                Channel channel= connection.createChannel();
        ) {
            channel.queueDeclare(RabbitmqUtils.QUEUE_NAME,true,false,false,null);
            channel.basicQos(64);
         //   basicConsume1(channel);
            basicConsume2(channel);
        }catch (Exception e) {
            log.error("consume message fail",e);
        }
    }

    /**
     * 消费方式一: 自动应答消息
     * @param channel 信道
     * @throws IOException IOException
     */
    private static void basicConsume1(Channel channel) throws IOException {
        DeliverCallback deliverCallback = getDeliverCallback();
        CancelCallback cancelCallback = getCancelCallback();
        channel.basicConsume(RabbitmqUtils.QUEUE_NAME,true,deliverCallback,cancelCallback);
    }

    /**
     * 消费方式二：手动应答消息
     * @param channel channel
     * @throws IOException IOException
     */
    private static void basicConsume2(Channel channel) throws Exception {
        // 使用 Consumer进行消费
        Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body)
                    throws IOException
            {
               log.info("receive message:{}",new String(body,StandardCharsets.UTF_8));

               channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        channel.basicConsume(RabbitmqUtils.QUEUE_NAME,consumer);
        // TODO 这里必须休眠，等待回调函数执行完毕后在关闭资源，否则抛出以下异常：
        //  AlreadyClosedException: channel is already closed due to clean channel shutdown
       TimeUnit.SECONDS.sleep(3);
    }

    /**
     * 取消消息的回调
     * @return 取消回调
     */
    private static CancelCallback getCancelCallback() {
        // 取消接收消息的回调
        return (consumerTag) -> {
          log.info("cancel callback:{}",consumerTag);
        };
    }

    /**
     * 发送成功的回调
     * @return 发送成功的回调
     */
    private static DeliverCallback getDeliverCallback() {
        // 接收到消息的回调
        return (consumerTag, message)-> {
            String textMessage = new String(message.getBody(), StandardCharsets.UTF_8);
            log.info("consumerTag:{}\n receive message:{}",consumerTag,textMessage);
        };
    }
}
